#include "BTreeNode.h"

using namespace std;

BTLeafNode::BTLeafNode()
{
    memset(leaf_pf.buffer, 0, sizeof(leaf_pf.buffer));
}

/*
 * Read the content of the node from the page pid in the PageFile pf.
 * @param pid[IN] the PageId to read
 * @param pf[IN] PageFile to read from
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTLeafNode::read(PageId pid, const PageFile& pf)
{
    // Read the contents of the leaf node buffer to the page file
    if(pf.read(pid, leaf_pf.buffer) != 0)
    {
        // @ERROR: Could not read the pagefile
        
        return(RC_FILE_READ_FAILED);
    }
    
    return(0);
}
    
/*
 * Write the content of the node to the page pid in the PageFile pf.
 * @param pid[IN] the PageId to write to
 * @param pf[IN] PageFile to write to
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTLeafNode::write(PageId pid, PageFile& pf)
{
    // Write the contents of the leaf node buffer to the page file
    if(pf.write(pid, leaf_pf.buffer) != 0)
    {
        // @ERROR: Could not write the leaf node contents to the pagefile
        
        return(RC_FILE_WRITE_FAILED);
    }
    
    return(0);
}

/*
 * Return the number of keys stored in the node.
 * @return the number of keys in the node
 */
int BTLeafNode::getKeyCount()
{
    return(leaf_pf.node.entryCount);
}

/*
 * Insert a (key, rid) pair to the node.
 * @param key[IN] the key to insert
 * @param rid[IN] the RecordId to insert
 * @return 0 if successful. Return an error code if the node is full.
 */
RC BTLeafNode::insert(int key, const RecordId& rid)
{
    if(leaf_pf.node.entryCount == MAX_NODE_ENTRIES)
    {
        // @ERROR: The node is full
        return(RC_NODE_FULL);
    }
    
    int eid;
    unsigned int entryCursor = leaf_pf.node.entryCount;
    
    // Finds the leaf node entry to insert the key
    locate(key, eid);
    
    // Shifts all the entries that follow the new entry one entry back in order 
    // to make space for the new entry
    while(entryCursor > eid)
    {
        leaf_pf.node.entries[entryCursor] = 
                leaf_pf.node.entries[entryCursor - 1];
        entryCursor--;
    }
    
    // Assigns the key to the correct entry in the leaf node
    leaf_pf.node.entries[eid].key = key;
    leaf_pf.node.entries[eid].rid = rid;
    
    leaf_pf.node.entryCount++;
    
    return(0);
}

/*
 * Insert the (key, rid) pair to the node
 * and split the node half and half with sibling.
 * The first key of the sibling node is returned in siblingKey.
 * @param key[IN] the key to insert.
 * @param rid[IN] the RecordId to insert.
 * @param sibling[IN] the sibling node to split with. This node MUST be EMPTY when this function is called.
 * @param siblingKey[OUT] the first key in the sibling node after split.
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTLeafNode::insertAndSplit(int key, const RecordId& rid, 
                              BTLeafNode& sibling, int& siblingKey)
{
    int retValue;
    
    if(sibling.leaf_pf.node.entryCount != 0)
    {
        // @ERROR: The sibling node is not empty
        
        return(RC_NODE_NOT_EMPTY);
    }
    else if(leaf_pf.node.entryCount < 2)
    {
        // @ERROR: Node does not have enough entries to be split
        
        return(RC_NODE_INSUF_ENTRIES);
    }
    else if(leaf_pf.node.entryCount == 1)
    {
        // The current node contains one entry
        
        if(key < leaf_pf.node.entries[0].key)
        {
            // The insertion key is less than the entry key value
            
            // Copy the entries in the nonleaf node to the sibling node
            memcpy(sibling.leaf_pf.node.entries, &leaf_pf.node.entries[0], 
                   sizeof(LeafNodeEntry));
            
            leaf_pf.node.entryCount = 0;
            sibling.leaf_pf.node.entryCount = 1;
            
            // Insert the entry into the current node
            retValue = insert(key, rid);
        }
        else
        {
            // The insertion key is greater than or equal to the entry key value
            
            // Insert the entry into the sibling node
            retValue = sibling.insert(key, rid);
        }
        
        return(retValue);
    }
    
    // Finds the initial split entry position in the current node
    unsigned int splitEntry = leaf_pf.node.entryCount / 2;
    
    // Determine in which node the inserted entry would reside
    if(key <= leaf_pf.node.entries[splitEntry].key)
    {   
        // The new entry belongs in the current node
        
        // Calculates the entry count in the sibling node
        if(leaf_pf.node.entryCount % 2 == 0)
        {
            sibling.leaf_pf.node.entryCount = splitEntry;
        }
        else
        {
            sibling.leaf_pf.node.entryCount = splitEntry + 1;
        }
        
        // Sets the entry count for the current node
        leaf_pf.node.entryCount = splitEntry;
        
        // Transfers the entries from the current node to the sibling node
        memcpy(sibling.leaf_pf.node.entries, 
               &leaf_pf.node.entries[splitEntry], 
               sibling.leaf_pf.node.entryCount * sizeof(LeafNodeEntry));
        
        // Inserts the entry into the current leaf node
        retValue = insert(key, rid);
    }
    else
    {
        // The new entry belongs in the sibling node
        
        // Calculates the entry count in the sibling node
        if(leaf_pf.node.entryCount % 2 == 0)
        {
            sibling.leaf_pf.node.entryCount = splitEntry - 1;
        }
        else
        {
            sibling.leaf_pf.node.entryCount = splitEntry;
        }
        
        if(leaf_pf.node.entryCount == 2)
        {
            // There are only two entries in the leaf node before the split
            
            leaf_pf.node.entryCount = splitEntry;
        
            // Transfers the entries from the current node to the sibling node
            memcpy(sibling.leaf_pf.node.entries, 
                   &leaf_pf.node.entries[splitEntry], sizeof(LeafNodeEntry));
        }
        else
        {
            // There are more than 3 entries in the leaf node before the split
            
            leaf_pf.node.entryCount = splitEntry + 1;
            
            // Transfers the entries from the current node to the sibling node
            memcpy(sibling.leaf_pf.node.entries, 
                   &leaf_pf.node.entries[splitEntry + 1], 
                   sibling.leaf_pf.node.entryCount * sizeof(LeafNodeEntry));
        }
        
        // Inserts the entry into the sibling leaf node
        retValue = sibling.insert(key, rid);
    }
    
    return(retValue);
}

/*
 * Find the entry whose key value is larger than or equal to searchKey
 * and output the eid (entry number) whose key value >= searchKey.
 * Remeber that all keys inside a B+tree node should be kept sorted.
 * @param searchKey[IN] the key to search for
 * @param eid[OUT] the entry number that contains a key larger than or equalty to searchKey
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTLeafNode::locate(int searchKey, int& eid)
{
    eid = 0;
    
    while(eid < leaf_pf.node.entryCount)
    {
        if(leaf_pf.node.entries[eid].key > searchKey)
        {
            return(0);
        }
        
        eid++;
    }
    
    return(0);
}

/*
 * Read the (key, rid) pair from the eid entry.
 * @param eid[IN] the entry number to read the (key, rid) pair from
 * @param key[OUT] the key from the entry
 * @param rid[OUT] the RecordId from the entry
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTLeafNode::readEntry(int eid, int& key, RecordId& rid)
{
    key = leaf_pf.node.entries[eid].key;
    rid = leaf_pf.node.entries[eid].rid;
    
    return(0);
}

/*
 * Return the pid of the next slibling node.
 * @return the PageId of the next sibling node 
 */
PageId BTLeafNode::getNextNodePtr()
{
    return(leaf_pf.node.nextPid);
}

/*
 * Set the pid of the next slibling node.
 * @param pid[IN] the PageId of the next sibling node 
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTLeafNode::setNextNodePtr(PageId pid)
{
    leaf_pf.node.nextPid = pid;
    
    return(0);
}

BTNonLeafNode::BTNonLeafNode()
{
    memset(nleaf_pf.buffer, 0, sizeof(nleaf_pf.buffer));
    nleaf_pf.node.overflowPid = NULL_PID;
}

/*
 * Read the content of the node from the page pid in the PageFile pf.
 * @param pid[IN] the PageId to read
 * @param pf[IN] PageFile to read from
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTNonLeafNode::read(PageId pid, const PageFile& pf)
{
    // Read the contents of the non-leaf node buffer to the page file
    if(pf.read(pid, nleaf_pf.buffer) != 0)
    {
        // @ERROR: Could not read the non-pagefile
        
        return(RC_FILE_READ_FAILED);
    }
    
    return(0);
}

/*
 * Write the content of the node to the page pid in the PageFile pf.
 * @param pid[IN] the PageId to write to
 * @param pf[IN] PageFile to write to
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTNonLeafNode::write(PageId pid, PageFile& pf)
{
    // Write the contents of the non-leaf node buffer to the page file
    if(pf.write(pid, nleaf_pf.buffer) != 0)
    {
        // @ERROR: Could not write the non-leaf node contents to the pagefile
        
        return(RC_FILE_WRITE_FAILED);
    }
    
    return(0);
}

/*
 * Return the number of keys stored in the node.
 * @return the number of keys in the node
 */
int BTNonLeafNode::getKeyCount()
{
    return(nleaf_pf.node.entryCount);
}

/*
 * Insert a (key, pid) pair to the node.
 * @param key[IN] the key to insert
 * @param pid[IN] the PageId to insert
 * @return 0 if successful. Return an error code if the node is full.
 */
RC BTNonLeafNode::insert(int key, PageId pid)
{
    if(nleaf_pf.node.entryCount == MAX_NODE_ENTRIES)
    {
        // @ERROR: The node is full
        
        return(RC_NODE_FULL);
    }
    
    PageId unused_pid;
    int eid;
    
    // Finds the entry position in the nonleaf node that the inserted entry 
    // would go
    if(locateChildPtr(key, unused_pid, eid) == RC_NODE_EMPTY)
    {
        // The node is empty
        
        nleaf_pf.node.entries[0].key = key;
        nleaf_pf.node.entries[0].pid = pid;
    }
    else if(eid == nleaf_pf.node.entryCount)
    {
        // The new entry belongs at the back of the nonleaf node
        
        nleaf_pf.node.entries[nleaf_pf.node.entryCount].key = key;
        nleaf_pf.node.entries[nleaf_pf.node.entryCount].pid = pid;
    }
    else
    {
        unsigned int entryCursor = nleaf_pf.node.entryCount;
        
        // Shifts all the entries that follow the new entry one entry back in 
        // order to make space for the new entry
        while(entryCursor > eid)
        {
            nleaf_pf.node.entries[entryCursor] = 
                    nleaf_pf.node.entries[entryCursor - 1];
            entryCursor--;
        }
        
        // Assigns the key to the correct entry in the nonleaf node
        nleaf_pf.node.entries[eid].key = key;
        nleaf_pf.node.entries[eid].pid = pid;
    }
    
    nleaf_pf.node.entryCount++;
    return(0);
}

/*
 * Insert the (key, pid) pair to the node
 * and split the node half and half with sibling.
 * The middle key after the split is returned in midKey.
 * @param key[IN] the key to insert
 * @param pid[IN] the PageId to insert
 * @param sibling[IN] the sibling node to split with. This node MUST be empty when this function is called.
 * @param midKey[OUT] the key in the middle after the split. This key should be inserted to the parent node.
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTNonLeafNode::insertAndSplit(int key, PageId pid, BTNonLeafNode& sibling, 
                                 int& midKey)
{
    int retValue;
    
    if(sibling.nleaf_pf.node.entryCount != 0)
    {
        // @ERROR: The sibling node is not empty
        
        return(RC_NODE_NOT_EMPTY);
    }
    else if(nleaf_pf.node.entryCount < 1)
    {
        // @ERROR: Node does not have enough entries to be split
        
        return(RC_NODE_INSUF_ENTRIES);
    }
    else if(nleaf_pf.node.entryCount == 1)
    {
        // The current node contains one entry
        
        if(key < nleaf_pf.node.entries[0].key)
        {
            // The insertion key is less than the entry key value
            
            // Copy the entries in the nonleaf node to the sibling node
            memcpy(sibling.nleaf_pf.node.entries, &nleaf_pf.node.entries[0], 
                   sizeof(NonLeafNodeEntry));
            
            nleaf_pf.node.entryCount = 0;
            sibling.nleaf_pf.node.entryCount = 1;
            
            // Insert the entry into the current node
            retValue = insert(key, pid);
            
            sibling.nleaf_pf.node.overflowPid = NULL_PID;
        }
        else
        {
            // The insertion key is greater than or equal to the entry key value
            
            // Insert the entry into the sibling node
            retValue = sibling.insert(key, pid);
            
            sibling.nleaf_pf.node.overflowPid = nleaf_pf.node.overflowPid;
            nleaf_pf.node.overflowPid = NULL_PID;
        }
        
        return(retValue);
    }
    
    // Finds the initial split entry position in the current node
    unsigned int splitEntry = nleaf_pf.node.entryCount / 2;
    
    // Determine in which node the inserted entry would reside
    if(key <= nleaf_pf.node.entries[splitEntry].key)
    {   
        // The new entry belongs in the current node
        
        // Calculates the entry count in the sibling node
        if(nleaf_pf.node.entryCount % 2 == 0)
        {
            sibling.nleaf_pf.node.entryCount = splitEntry;
        }
        else
        {
            sibling.nleaf_pf.node.entryCount = splitEntry + 1;
        }
        
        // Sets the entry count for the current node
        nleaf_pf.node.entryCount = splitEntry;
        
        // Transfers the entries from the current node to the sibling node
        memcpy(sibling.nleaf_pf.node.entries, 
               &nleaf_pf.node.entries[splitEntry], 
               sibling.nleaf_pf.node.entryCount * sizeof(NonLeafNodeEntry));
        
        // Inserts the entry into the current nonleaf node
        retValue = insert(key, pid);
    }
    else
    {
        // The new entry belongs in the sibling node
        
        // Calculates the entry count in the sibling node
        if(nleaf_pf.node.entryCount % 2 == 0)
        {
            sibling.nleaf_pf.node.entryCount = splitEntry - 1;
        }
        else
        {
            sibling.nleaf_pf.node.entryCount = splitEntry;
        }
        
        if(nleaf_pf.node.entryCount == 2)
        {
            // There are only two entries in the nonleaf node before the split
            
            nleaf_pf.node.entryCount = splitEntry;
        
            // Transfers the entries from the current node to the sibling node
            memcpy(sibling.nleaf_pf.node.entries, 
                   &nleaf_pf.node.entries[splitEntry], 
                   sizeof(NonLeafNodeEntry));
        }
        else
        {
            // There are more than 3 entries in the nonleaf node before the 
            // split
            
            nleaf_pf.node.entryCount = splitEntry + 1;
            
            // Transfers the entries from the current node to the sibling node
            memcpy(sibling.nleaf_pf.node.entries, 
                   &nleaf_pf.node.entries[splitEntry + 1], 
                   sibling.nleaf_pf.node.entryCount * sizeof(NonLeafNodeEntry));
        }
        
        // Inserts the entry into the sibling nonleaf node
        retValue = sibling.insert(key, pid);
    }
    
    sibling.nleaf_pf.node.overflowPid = nleaf_pf.node.overflowPid;
    nleaf_pf.node.overflowPid = NULL_PID;
    
    // Obtains the middle key from the parent node
    midKey = nleaf_pf.node.entries[nleaf_pf.node.entryCount - 1].key;
    
    return(retValue);
}

/*
 * Given the searchKey, find the child-node pointer to follow and
 * output it in pid.
 * @param searchKey[IN] the searchKey that is being looked up.
 * @param pid[OUT] the pointer to the child node to follow.
 * @param eid[OUT] the entry position of the key.
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTNonLeafNode::locateChildPtr(int searchKey, PageId& pid, int &eid)
{
    unsigned int entryCount = nleaf_pf.node.entryCount;
    
    if(entryCount <= 0)
    {
        // @ERROR: Non-leaf node is empty
        
        return(RC_NODE_EMPTY);
    }
    
    if(searchKey < nleaf_pf.node.entries[0].key)
    {
        // The search key is less than the smallest nonleaf entry key
        
        eid = 0;
        pid = nleaf_pf.node.entries[0].pid;
        
        return(0);
    }
    
    eid = 1;
    int prevKey = nleaf_pf.node.entries[0].key;
    int curKey;
    
    // Loop through all the entries in the nonleaf node
    while(eid < entryCount)
    {
        curKey = nleaf_pf.node.entries[eid].key;
        
        // Check if the search key is between the bounds of the current key 
        // and the previous key
        if((searchKey >= prevKey) && (searchKey < curKey))
        {
            pid = nleaf_pf.node.entries[eid].pid;
            return(0);
        }
        
        prevKey = curKey;
        eid++;
    }
    
    pid = nleaf_pf.node.overflowPid;
    
    return(0);
}

/*
 * Initialize the root node with (pid1, key, pid2).
 * @param pid1[IN] the first PageId to insert
 * @param key[IN] the key that should be inserted between the two PageIds
 * @param pid2[IN] the PageId to insert behind the key
 * @return 0 if successful. Return an error code if there is an error.
 */
RC BTNonLeafNode::initializeRoot(PageId pid1, int key, PageId pid2)
{
    nleaf_pf.node.entries[0].key = key;
    nleaf_pf.node.entries[0].pid = pid1;
    
    nleaf_pf.node.overflowPid = pid2;
    nleaf_pf.node.entryCount = 1;
    
    return(0);
}

void BTNonLeafNode::setOverflowPid(PageId pid)
{
    nleaf_pf.node.overflowPid = pid;
}
